package org.apache.pulsar.testclient;

import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

import static java.util.concurrent.TimeUnit.NANOSECONDS;

/**
 * @author zhangjg
 * @desc s
 * @date 2023/03/29 14:10
 **/
public class PTest {

    private static final Logger log = LoggerFactory.getLogger(PTest.class);

    public static void main(String[] args) throws Exception {

        ClientBuilder clientBuilder = PulsarClient.builder();
        clientBuilder.enableTransaction(false)
                .serviceUrl("http://127.0.0.1:8080")
                .memoryLimit(100, SizeUnit.MEGA_BYTES);

        PulsarClient client = clientBuilder.build();

        ProducerBuilder<byte[]> producerBuilder = client.newProducer()
                .batchingMaxMessages(100)
                .blockIfQueueFull(true)
                .topic("persistent://public/default/tt11");

        Producer<byte[]> producer = producerBuilder.create();

        String msg = "xxxxxx";

        producer.newMessage(Schema.BYTES)
                .value(msg.getBytes(StandardCharsets.UTF_8))
                .sendAsync().thenRun(() -> {
            log.info("Write message success. msg=", msg);
        }).exceptionally(ex -> {
            // Ignore the exception of recorder since a very large latencyMicros will lead
            if (ex.getCause() instanceof ArrayIndexOutOfBoundsException) {
                return null;
            }
            log.warn("Write message error with exception", ex);

            return null;
        });

        log.info("---------------------end");
        client.close();

    }
}
